package com.bubbles.workspace.sync.service.source.metadata;


import com.baomidou.dynamic.datasource.annotation.DS;
import com.bubbles.engine.data.core.repository.BaseRepository;
import com.bubbles.engine.data.core.service.BaseService;
import com.bubbles.workspace.sync.entity.source.metadata.*;
import com.bubbles.workspace.sync.key.source.metadata.SourceColumnKey;
import com.bubbles.workspace.sync.properties.SshProperties;
import com.bubbles.workspace.sync.repository.source.metadata.SourceColumnRepository;
import com.bubbles.workspace.sync.service.source.constraint.*;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDimensionService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDorisAnalyzeColumnService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDorisAnalyzeTableService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceIncrementService;
import com.bubbles.workspace.sync.util.JdbcUtil;
import com.bubbles.workspace.sync.util.SshManager;
import com.google.common.collect.Maps;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>description: 源数据源-列 </p>
 *
 * @author CryptoNeedle
 * @date 2025-06-20
 */
@DS("pg")
@Slf4j
@Service
public class SourceColumnService extends BaseService<SourceColumn, SourceColumnKey> {
    
    @Resource
    private SourceColumnRepository sourceColumnRepository;
    
    @Override
    public BaseRepository<SourceColumn, SourceColumnKey> getRepository() {
        return this.sourceColumnRepository;
    }
    
    @Resource
    private SourceJdbcService sourceJdbcService;
    
    @Resource
    private SourceServerService sourceServerService;
    
    @Resource
    private SourceCatalogService sourceCatalogService;
    
    @Resource
    private SourceDatabaseService sourceDatabaseService;
    
    @Resource
    private SourceTableService sourceTableService;
    
    @Resource
    private SourceColumnService sourceColumnService;
    
    @Resource
    private SourceConstraintPrimaryService sourceConstraintPrimaryService;
    
    @Resource
    private SourceConstraintUniqueService sourceConstraintUniqueService;
    
    @Resource
    private SourceConstraintForeignService sourceConstraintForeignService;
    
    @Resource
    private SourceConstraintCheckService sourceConstraintCheckService;
    
    @Resource
    private SourceIndexUniqueService sourceIndexUniqueService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeTableService sourceEnhanceDorisAnalyzeTableService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeColumnService sourceEnhanceDorisAnalyzeColumnService;
    
    @Resource
    private SourceEnhanceIncrementService sourceEnhanceIncrementService;
    
    @Resource
    private SourceEnhanceDimensionService sourceEnhanceDimensionService;
    
    @Resource
    private SshProperties sshProperties;
    
    @Resource
    private SshManager sshManager;
    
    @Transactional
    public List<SourceColumn> listByDatabase(SourceDatabase sourceDatabase) {
        return sourceColumnRepository.listByDatabase(sourceDatabase.getSourceCatalogName(), sourceDatabase.getSourceDatabaseName());
    }
    
    @Transactional
    public void addList(List<SourceColumn> sourceColumns) {
        sourceColumnService.saveAll(sourceColumns);
    }
    
    @Transactional
    public void saveList(List<SourceColumn> sourceColumns) {
        sourceColumnService.saveAll(sourceColumns);
    }
    
    @Transactional
    public void deleteList(List<SourceColumn> sourceColumns) {
        sourceColumnService.deleteAll(sourceColumns);
    }
    
    @Transactional
    public void syncSource(SourceCatalog sourceCatalog, SourceDatabase sourceDatabase) {
        String sourceCatalogName = sourceCatalog.getSourceCatalogName();
        String sourceDatabaseName = sourceDatabase.getSourceDatabaseName();
        SourceJdbc sourceJdbc = sourceCatalog.getSourceJdbc();
        List<SourceTable> sourceTables = sourceTableService.listByDatabase(sourceDatabase);
        Map<String, SourceTable> sourceTableMap = Maps.uniqueIndex(sourceTables, SourceTable::getSourceTableName);
        
        JdbcTemplate jdbcTemplate = JdbcUtil.getJdbcTemplate(sourceJdbc);
        List<SourceColumn> sourceColumns = jdbcTemplate.query(sourceJdbc
                                                                      .getDatabaseType()
                                                                      .getDatabase()
                                                                      .getColumnSql(sourceDatabaseName), (resultSet, rowNum) -> new SourceColumn()
                .setSourceCatalogName(sourceCatalogName)
                .setSourceSystemCode(sourceDatabase.getSourceSystemCode())
                .setSourceDatabaseName(sourceDatabaseName)
                .setSourceTableName(resultSet.getString("sourceTableName"))
                .setSourceColumnName(resultSet.getString("sourceColumnName"))
                .setSourceColumnComment(resultSet.getString("sourceColumnComment"))
                .setSourceColumnCommentManual(false)
                .setSourceDataType(resultSet.getString("sourceDataType"))
                .setSourceLength(resultSet.getLong("sourceLength"))
                .setSourcePrecision(resultSet.getLong("sourcePrecision"))
                .setSourceScale(resultSet.getLong("sourceScale"))
                .setSourcePosition(resultSet.getLong("sourcePosition"))
                .setSourceStatistic(new SourceColumn.SourceStatistic()
                                            .setSourceStatisticRowCount(sourceTableMap.get(resultSet.getString("sourceTableName")) != null ?
                                                                        sourceTableMap
                                                                                .get(resultSet.getString("sourceTableName"))
                                                                                .getSourceStatistic()
                                                                                .getSourceStatisticRowCount() : null)
                                            .setSourceStatisticSampleCount(resultSet.getLong("sourceStatisticSampleCount"))
                                            .setSourceStatisticNullCount(resultSet.getLong("sourceStatisticNullCount"))
                                            .setSourceStatisticDistinctCount(resultSet.getLong("sourceStatisticDistinctCount"))
                                            .setSourceStatisticDensity(resultSet.getBigDecimal("sourceStatisticDensity"))
                                            .setSourceStatisticMinValue(resultSet.getString("sourceStatisticMinValue"))
                                            .setSourceStatisticMaxValue(resultSet.getString("sourceStatisticMaxValue"))
                                            .setSourceStatisticAvgDataSizeByte(resultSet.getBigDecimal("sourceStatisticAvgDataSizeByte"))
                                            .setSourceStatisticDataSize(null)
                                            .setSourceStatisticDatetime(resultSet.getDate("sourceStatisticTime")))
                .setDorisStatistic(null)
                .setDetailStatistic(null)
                .setDoris(null)
                .setSourceTable(sourceTableMap.get(resultSet.getString("sourceTableName")))
                .setSourceConstraintPrimaries(null)
                .setSourceConstraintUniques(null)
                .setSourceConstraintForeigns(null)
                .setSourceIndexUniques(null)
                .setSourceEnhanceDimensions(null)
                .setSourceEnhanceDorisAnalyzeColumns(null));
        
        sourceColumnService.compareSource(sourceDatabase, sourceColumns);
    }
    
    @Transactional
    public void compareSource(SourceDatabase sourceDatabase, List<SourceColumn> dealList) {
        List<SourceColumn> originList = sourceColumnService.listByDatabase(sourceDatabase);
        // 已存在
        Map<String, SourceColumn> originMap = Maps.uniqueIndex(originList, origin -> origin.getStringKey());
        // 待处理
        Map<String, SourceColumn> dealMap = Maps.uniqueIndex(dealList, deal -> deal.getStringKey());
        
        // 新增
        List<SourceColumn> extraList = dealList.stream().filter(deal -> !originMap.containsKey(deal.getStringKey())).toList();
        
        // 保存
        List<SourceColumn> existsList = originList.stream().filter(origin -> dealMap.containsKey(origin.getStringKey())).peek(origin -> {
            SourceColumn deal = dealMap.get(origin.getStringKey());
            if (!origin.getSourceColumnCommentManual()) {
                origin.setSourceColumnComment(deal.getSourceColumnComment());
            }
            origin
                    .setSourceDataType(deal.getSourceDataType())
                    .setSourceLength(deal.getSourceLength())
                    .setSourcePrecision(deal.getSourcePrecision())
                    .setSourceScale(deal.getSourceScale())
                    .setSourcePosition(deal.getSourcePosition());
            origin.setSourceStatistic(deal.getSourceStatistic());
        }).toList();
        
        // 移除
        List<SourceColumn> missList = originList.stream().filter(item -> !dealMap.containsKey(item.getStringKey())).toList();
        
        sourceColumnService.addList(extraList);
        sourceColumnService.saveList(existsList);
        sourceColumnService.deleteList(missList);
        
        log.info("[SYNC] {} -> {} -> 元数据采集 -> Column             => 共计{} | 新增{} | 删除{}", sourceDatabase.getSourceCatalogName(), sourceDatabase.getSourceDatabaseName(), String.format("%5d", dealList.size()), String.format("%5d", extraList.size()), String.format("%5d", missList.size()) + (
                !missList.isEmpty() ? " => " + missList
                        .stream()
                        .map(sourceColumn -> sourceColumn.getSourceTableName() + " - " + sourceColumn.getSourceColumnName())
                        .collect(Collectors.joining(",")) : ""));
    }
}